
package com.hexinfo.dmpro.sparing.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.date.DateField;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.file.FileAppender;
import cn.hutool.core.text.StrSpliter;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hexinfo.dmpro.common.model.ClusterSparingConf;
import com.hexinfo.dmpro.common.utils.CommonConstants;
import com.hexinfo.dmpro.common.utils.ScanCommonConstants;
import com.hexinfo.dmpro.common.utils.StringUtils;
import com.hexinfo.dmpro.sparing.async.AsyncTask;
import com.hexinfo.dmpro.sparing.dto.*;
import com.hexinfo.dmpro.sparing.model.*;
import com.hexinfo.dmpro.sparing.dao.SyncRuleConfMapper;
import com.hexinfo.dmpro.sparing.service.*;
import lombok.AllArgsConstructor;


import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.File;
import java.util.*;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

/**
 * 金桥到万国同步(源级)
 *
 * @author yemw
 * @date 2023-09-08 17:45:29
 */
@Service
@Slf4j
public class SyncRuleConfServiceImpl extends ServiceImpl<SyncRuleConfMapper, SyncRuleConf> implements SyncRuleConfService {
    @Value("${hdfs.tarPrefix}")
    private String prefix;
    @Autowired
    private ScanMetadataService scanMetadataService;
    @Autowired
    private TimeFieldMatchingService timeFieldMatchingService;
    @Autowired
    private HiveToOracleService hiveToOracleService;
    @Autowired
    private RuleToDistCPService ruleToDistCPService;
    @Autowired
    private AvailabilityPathService availabilityPathService;
    @Autowired
    private HotStandbyLogSourceService hotStandbyLogSourceService;
    @Autowired
    private SyncRuleConfMapper syncRuleConfMapper;
    @Autowired
    private AsyncTask asyncTask;
    @Autowired
    private XxJobLogService xxJobLogService;
    @Autowired
    private SyncCompleteLibraryTaskService syncCompleteLibraryTaskService;

    @Override
    public Page queryList(Page page, SyncRuleConf syncRuleConf) {
        LambdaQueryWrapper<SyncRuleConf> wrapper = Wrappers.lambdaQuery();
        wrapper.like(StrUtil.isNotBlank(syncRuleConf.getSyncRuleName()),SyncRuleConf::getSyncRuleName,syncRuleConf.getSyncRuleName())
                .like(StrUtil.isNotBlank(syncRuleConf.getSourceName())&&!"无源".equals(syncRuleConf.getSourceName()),SyncRuleConf::getSourceName, StringUtils.toUpperCase(syncRuleConf.getSourceName()))
                .isNull("无源".equals(syncRuleConf.getSourceName()),SyncRuleConf::getSourceName)
                .eq(StrUtil.isNotBlank(syncRuleConf.getIsPart()),SyncRuleConf::getIsPart,syncRuleConf.getIsPart())
                .eq(StrUtil.isNotBlank(syncRuleConf.getIsTimePart()),SyncRuleConf::getIsTimePart,syncRuleConf.getIsTimePart())
                .eq(StrUtil.isNotBlank(syncRuleConf.getSyncMode()),SyncRuleConf::getSyncMode,syncRuleConf.getSyncMode())
                .eq(SyncRuleConf::getDel,CommonConstants.STATUS_NORMAL)
                .orderByDesc(SyncRuleConf::getSourceName)
                ;
        Page returnPage = this.baseMapper.selectPage(page, wrapper);
        return returnPage;
    }

    /**
     *
     * @param  flag 数据、文档
     * @param source 数据源标识
     * @param cpWhether 是否distCP
     * @return
     */
    @Override
    public Boolean qyScanMetadataWhere(String flag,String source,String cpWhether) {
        //新增日志分析记录--获取id
        String logId = hotStandbyLogSourceService.addHotStandbyLogSource(source);
//        if (StrUtil.isBlank(logId)) {
//            log.error("-------------------新增日志分析记录异常");
//            return false;
//        }
//
//        String nameNodeStatus = availabilityPathService.getSouAndTar();
//        if (StrUtil.isBlank(nameNodeStatus)){
//            log.error("-------------------高可用节点变动异常");
//            return false;
//        }
        String id = null;
        boolean bl = false;
        if(StrUtil.isNotBlank(flag)){
            if("data".equals(flag)){
                //判断是否第一次同步
                int count = scanMetadataService.count();
                if(count>0){
                    id = scanMetadataService.hiveSqlOracleIncrement(source);
                }else{
                    id = scanMetadataService.hiveMetadataToOracle(source);
                }
                bl = getSyncRuleData(source,cpWhether,id,logId);
            }
            else if("word".equals(flag)){
                getSyncRuleDoc(source);
            }
        }
        return bl;
    }

    /**
     * 需要hive中新建的表信息
     * @return
     */
    @Override
    public List<CreateTableDTO> createTable(String dbName) {
        Date date = syncCompleteLibraryTaskService.maxCreateTime(dbName);
        long staTime = date.getTime() / 1000;
        long endTime = System.currentTimeMillis() / 1000;
        return hiveToOracleService.selectHiveLatestTable(dbName,staTime,endTime);
    }



    /**
     * 获取第几个斜杆所在位置
     * @return
     */
    public int getSlashIndex(String url,int num){
        int slashCount = 0; // 记录斜杆的数量
        int slashIndex = -1; // 记录第几个斜杆的位置
        for (int i = 0; i < url.length(); i++) {
            if (url.charAt(i) == '/') {
                slashCount++;
                if (slashCount == num) {
                    slashIndex = i;
                    break;
                }
            }
        }
        return slashIndex;
    }

    /**
     * 通过同步规则(数据)distCP
     * @param source 源标识
     * @param cpWhether 是否distCP
     * @return
     */
    @Override
    public Boolean getSyncRuleData(String source,String cpWhether,String id,String logId) {
        //查询规则
        LambdaQueryWrapper<SyncRuleConf> syncRuleWrapper = Wrappers.lambdaQuery();
        syncRuleWrapper.eq(SyncRuleConf::getDel, CommonConstants.STATUS_NORMAL)
//                .eq(SyncRuleConf::getExeFrequency, ScanCommonConstants.ClusterName.ONE.value)
                .eq(StrUtil.isNotBlank(source)&&!"dmpro".equals(source),SyncRuleConf::getSourceName,source)
                .isNull(StrUtil.isBlank(source),SyncRuleConf::getSourceName);
        List<SyncRuleConf> syncRuleConfs = this.baseMapper.selectList(syncRuleWrapper);
        //查询集群信息
        LambdaQueryWrapper<ClusterSparingConf> jQWrapper = Wrappers.lambdaQuery();
        LambdaQueryWrapper<ClusterSparingConf> wGWrapper = Wrappers.lambdaQuery();
        ClusterSparingConf jQCluster = scanMetadataService.wrapperClusterSparingConf(jQWrapper,"金桥");
        ClusterSparingConf wGCluster = scanMetadataService.wrapperClusterSparingConf(wGWrapper,"万国");
        //DistCP集合
        List<HdfsHeatBackup> hdfsHeatBackups = new ArrayList<>();
        //对比条件集合
        List<WhereScriptDTO> whereScriptDTOS = new ArrayList<>();
        RuleToDistCPDTO ruleToDistCPDTO = new RuleToDistCPDTO();
        Set<String> syncRuleConfSet = new HashSet<>();
        syncRuleConfs.forEach(syncRule -> {
            ScanSyncRuleDTO syncRuleDTO = splQuSoeDataTable(syncRule);
            if(StrUtil.isNotBlank(syncRule.getPartValueWhite())){
                List<String> set = Arrays.asList(syncRule.getPartValueWhite().split(","));
                if(ObjectUtil.isNotEmpty(set)){
                    for (String str: set) {
                        syncRuleConfSet.add(str);
                    }
                }
            }

            //判断同步范围 全量还是近2年 0为全量
            //全量数据不需要判断时间分区
            if(syncRule.getSyncRange()==0){
                log.warn("================当前规则为:"+syncRule.getSyncRuleName());
                List<ScanMetadata> scanMetadataList = scanMetadataService.qyScanMetadataWhereWord(syncRuleDTO);
                scanMetadataList.forEach(s ->{
                    log.warn("==========当前表为:"+s.getTblName());
                    if("源标识同步".equals(syncRule.getSyncMode())&&"分区".equals(syncRule.getIsPart())&&"无时间分区".equals(syncRule.getIsTimePart())){
                        //拼接条件
                        String where = whereLike(syncRule);
                        //查询该表源的hdfs路径
                        List<String> hdfsList = hiveToOracleService.selectHiveLocation(where,s.getTblName(),s.getTblDatabaseName());
                        if(hdfsList!=null&&hdfsList.size()>0){
                            for (String hdfs: hdfsList) {
                                HdfsHeatBackup kup = new HdfsHeatBackup();
                                WhereScriptDTO whereScriptDTO = new WhereScriptDTO();
                                String lo = hdfs.substring(getSlashIndex(hdfs,6) + 1);
                                kup.setSourceAddress(jQCluster.getHdfsPrefixPath()+lo);
                                kup.setTargetAddress(wGCluster.getHdfsPrefixPath()+prefix+lo);
                                kup.setSourceDbName(s.getTblDatabaseName());
                                kup.setTargetDbName(prefix+s.getTblDatabaseName());
                                kup.setSourceTableName(s.getTblName());
                                kup.setTargetTableName(s.getTblName());
                                kup.setBackupSort("1");
                                ruleToDistCPDTO.setBackupSort("1");
                                whereScriptDTO.setSourceDbName(s.getTblDatabaseName());
                                whereScriptDTO.setSourceTableName(s.getTblName());
                                kup.setSourceData(syncRule.getSourceName());
                                ruleToDistCPDTO.setSourceDataName(syncRule.getSourceName());
                                String[] partNames = hdfs.substring(getSlashIndex(hdfs,8) + 1).split("=");
                                whereScriptDTO.setSourceQueryCriteria(StrUtil.replace(StrUtil.replace(hdfs.substring(getSlashIndex(hdfs,8) + 1),"=","='"),partNames[0],"cast( "+ partNames[0] +" as string )") + "'");
                                whereScriptDTO.setTargetQueryCriteria(whereScriptDTO.getSourceQueryCriteria());
                                whereScriptDTO.setTargetTableName(s.getTblName());
                                whereScriptDTO.setTargetDbName(prefix+s.getTblDatabaseName());
                                whereScriptDTOS.add(whereScriptDTO);
                                hdfsHeatBackups.add(kup);
                                log.warn("==========================【全量源标识】当前数据比对条件"+whereScriptDTO.getSourceQueryCriteria());
                            }
                        }else{
                            whereJH(jQCluster,wGCluster,s,prefix,syncRule,ruleToDistCPDTO,whereScriptDTOS,syncRuleConfSet,hdfsHeatBackups);
                        }
                    }else {
                        WhereScriptDTO whereScriptDTO = new WhereScriptDTO();
                        HdfsHeatBackup kup = new HdfsHeatBackup();
                        if("分区".equals(syncRule.getIsPart())&&"特殊表含时间分区".equals(syncRule.getIsTimePart())){
                            kup.setSourceAddress(jQCluster.getHdfsPrefixPath()+s.getTblDatabaseName()+".db/"+s.getTblName()+"/"+syncRule.getPartValueWhite());
                            kup.setTargetAddress(wGCluster.getHdfsPrefixPath()+prefix+s.getTblDatabaseName()+".db/"+s.getTblName()+"/"+syncRule.getPartValueWhite());
                            String[] partNames = syncRule.getPartValueWhite().split("=");
                            //当有时间分区时对比条件拼接要加上时间分区
                            if(StrUtil.isBlank(syncRule.getPartNameWhite())){
                                //无时间分区
                                whereScriptDTO.setSourceQueryCriteria(StrUtil.replace(StrUtil.replace(syncRule.getPartValueWhite(),"=","='"),partNames[0],"cast( "+ partNames[0] +" as string )") + "'");
                            }else{
                                //有时间分区
                                LambdaQueryWrapper<TimeFieldMatching> timeFieldWrapper = Wrappers.lambdaQuery();
                                timeFieldWrapper.select(TimeFieldMatching::getTimeFieldKay, TimeFieldMatching::getTimeFieldValue);
                                timeFieldWrapper.eq(TimeFieldMatching::getTimeFieldKay, syncRule.getPartNameWhite())
                                                .eq(TimeFieldMatching::getDel, CommonConstants.STATUS_NORMAL);
                                TimeFieldMatching matching = timeFieldMatchingService.getOne(timeFieldWrapper);
//                                String endTime = DateUtil.format(DateUtil.date(), matching.getTimeFieldValue());
                                int endDetweenDay = 15;
                                String endTime = DateUtil.format(DateUtil.offset(DateUtil.date(), DateField.DAY_OF_MONTH, endDetweenDay), matching.getTimeFieldValue());
                                int betweenDay = -7500;
                                String staTime = DateUtil.format(DateUtil.offset(DateUtil.date(), DateField.DAY_OF_MONTH, betweenDay), matching.getTimeFieldValue());
                                whereScriptDTO.setSourceQueryCriteria(StrUtil.replace(StrUtil.replace(syncRule.getPartValueWhite(),"=","='"),partNames[0],"cast( "+ partNames[0] +" as string )") + "' and "+"cast( "+ syncRule.getPartNameWhite() +" as int )"+ " BETWEEN " + staTime +" and "+ endTime);
                            }
                            whereScriptDTO.setTargetQueryCriteria(whereScriptDTO.getSourceQueryCriteria());
                        }else {
                            kup.setSourceAddress(jQCluster.getHdfsPrefixPath()+s.getTblDatabaseName()+".db/"+s.getTblName());
                            kup.setTargetAddress(wGCluster.getHdfsPrefixPath()+prefix+s.getTblDatabaseName()+".db/"+s.getTblName());
                        }
                        kup.setSourceDbName(s.getTblDatabaseName());
                        kup.setTargetDbName(prefix+s.getTblDatabaseName());
                        kup.setSourceTableName(s.getTblName());
                        kup.setTargetTableName(s.getTblName());
                        whereScriptDTO.setSourceDbName(s.getTblDatabaseName());
                        whereScriptDTO.setSourceTableName(s.getTblName());
                        if(StrUtil.isNotBlank(syncRule.getSourceName())){
                            kup.setSourceData(syncRule.getSourceName());
                            ruleToDistCPDTO.setSourceDataName(syncRule.getSourceName());
                            kup.setBackupSort("1");
                            ruleToDistCPDTO.setBackupSort("1");
                        }else{
                            kup.setSourceData("无源");
                            ruleToDistCPDTO.setSourceDataName("无源");
                            kup.setBackupSort("1");
                            ruleToDistCPDTO.setBackupSort("1");
                        }
                        if(StrUtil.isBlank(whereScriptDTO.getSourceQueryCriteria())){
                            whereScriptDTO.setSourceQueryCriteria("");
                            whereScriptDTO.setTargetQueryCriteria(whereScriptDTO.getSourceQueryCriteria());
                        }
                        whereScriptDTO.setTargetTableName(s.getTblName());
                        whereScriptDTO.setTargetDbName(prefix+s.getTblDatabaseName());
                        whereScriptDTOS.add(whereScriptDTO);
                        log.warn("==========================【常规/特殊】当前数据比对条件"+whereScriptDTO.getSourceQueryCriteria());
                        hdfsHeatBackups.add(kup);
                    }
                });
                try {
                    asyncTask.updateToSyncType(syncRuleDTO);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                //近2年/近几年数据一定包含时间分区
            }else if(syncRule.getSyncRange()>0){
                int betweenDay = syncRule.getSyncRange();
                log.warn("================当前规则为:"+syncRule.getSyncRuleName());
                //分组拿表名
                List<ScanMetadata> scanMetadataList = scanMetadataService.qyScanMetadataWhereWord(syncRuleDTO);
                scanMetadataList.forEach(s ->{
                    log.warn("==========当前表为:"+s.getTblName());
                    //查询该表的全部分区字段
                    List<String> hdfsList = hiveToOracleService.selectHivePkeyName(s.getTblName(),s.getTblDatabaseName());
                    //循环匹配分区字段配置表 拿到字段名称和字段格式
                    if(hdfsList!=null&&hdfsList.size()>0){
                        for (String h : hdfsList) {
                            LambdaQueryWrapper<TimeFieldMatching> timeFieldWrapper = Wrappers.lambdaQuery();
                            timeFieldWrapper.select(TimeFieldMatching::getTimeFieldKay, TimeFieldMatching::getTimeFieldValue);
                            timeFieldWrapper.eq(TimeFieldMatching::getTimeFieldKay, h).eq(TimeFieldMatching::getDel, CommonConstants.STATUS_NORMAL);
                            TimeFieldMatching matching = timeFieldMatchingService.getOne(timeFieldWrapper);
                            //不为空就匹配成功
                            if (ObjectUtils.isNotEmpty(matching)) {
                                //拼接条件
                                String where = whereLike(syncRule);
                                //获取当前规则源2年分区的hdfs地址
                                List<String> locationList = hiveToOracleService.selectHiveTableHDFS(s.getTblName(), where, matching.getTimeFieldKay(), -betweenDay, matching.getTimeFieldValue(), s.getTblDatabaseName());
                                //拼接数据对比所需条件 出现类似于订单TA源分区多种value值
                                WhereScriptDTO whereScriptDTO =  new WhereScriptDTO();
                                if(ObjectUtil.isNotEmpty(locationList)){
                                    whereScriptDTO  =  whereScript(locationList, s.getTblDatabaseName(), s.getTblName(), matching, syncRule.getSyncMode(), -betweenDay, prefix);
                                    Set<kvDTO> locationGorup = new HashSet<>();
                                    locationList.forEach(lo -> {
                                        HdfsHeatBackup kup = new HdfsHeatBackup();
                                        //切割分区信息
                                        String partName = "";
                                        if (StrUtil.split(lo, s.getTblName()).length == 2) {
                                            partName = StrUtil.split(lo, s.getTblName())[1];
                                        } else {
                                            log.warn("===========" + lo + "===========");
                                        }
                                        //（为了避免该表有多个时间分区字段） 且不等于无源
                                        if(StrUtil.isNotBlank(partName)&&StrUtil.isNotBlank(syncRule.getSourceName())){
                                            kvDTO kvDTO = new kvDTO();
                                            //判断时间分区在前在后 切割源标识
                                            String[] names = partName.substring(1,partName.length()).split("/");
                                            String[] partNamesQ = names[0].split("=");
                                            if(names.length>=2){
                                                String[] partNamesH = names[1].split("=");
                                                //分区在前
                                                if(matching.getTimeFieldKay().equals(partNamesQ[0])){
                                                    kvDTO.setSourceKey(names[1]);
                                                    kvDTO.setSourceValue(partNamesH[0]);
                                                    kvDTO.setTimeKey(partNamesQ[0]);
                                                    locationGorup.add(kvDTO);
                                                    //分区在后
                                                }else if(matching.getTimeFieldKay().equals(partNamesH[0])){
                                                    kvDTO.setSourceKey(names[0]);
                                                    kvDTO.setSourceValue(partNamesQ[0]);
                                                    kvDTO.setTimeKey(partNamesH[0]);
                                                    locationGorup.add(kvDTO);
                                                }
                                            }else{
                                                kvDTO.setSourceKey(names[0]);
//                                        kvDTO.setSourceValue(partNamesQ[0]);
                                                locationGorup.add(kvDTO);
                                            }
                                        }
                                        lo = jQCluster.getHdfsPrefixPath() + s.getTblDatabaseName() + ".db/" + s.getTblName() + partName;
                                        kup.setSourceAddress(lo);
                                        kup.setTargetAddress(StrUtil.replace(StrUtil.replace(lo, jQCluster.getHdfsPrefixPath(), wGCluster.getHdfsPrefixPath()), "/" + s.getTblDatabaseName() + ".db/", "/" + prefix + s.getTblDatabaseName() + ".db/"));
                                        kup.setSourceDbName(s.getTblDatabaseName());
                                        kup.setTargetDbName(prefix + s.getTblDatabaseName());
                                        kup.setSourceTableName(s.getTblName());
                                        kup.setTargetTableName(s.getTblName());
                                        kup.setPkeyName(matching.getTimeFieldKay());
                                        kup.setSyncRange(syncRule.getSyncRange());
                                        if (StrUtil.isNotBlank(syncRule.getSourceName())) {
                                            kup.setSourceData (syncRule.getSourceName());
                                            ruleToDistCPDTO.setSourceDataName(syncRule.getSourceName());
                                            kup.setBackupSort("1");
                                            ruleToDistCPDTO.setBackupSort("1");
                                        } else {
                                            kup.setSourceData("无源");
                                            ruleToDistCPDTO.setSourceDataName("无源");
                                            kup.setBackupSort("1");
                                            ruleToDistCPDTO.setBackupSort("1");
                                        }
                                        hdfsHeatBackups.add(kup);
                                    });
                                    //源分区单种value值（为了避免该表有多个时间分区字段） 或者无源
                                    if((locationGorup!=null&&locationGorup.size()==1)||StrUtil.isBlank(syncRule.getSourceName())){
                                        log.warn("==========================【源分区单种value值】当前数据比对条件"+whereScriptDTO.getSourceQueryCriteria());
                                        whereScriptDTOS.add(whereScriptDTO);
                                        break;
                                        //源分区多种value值
                                    }else if(locationGorup!=null&&locationGorup.size()>1){
                                        for (kvDTO kv : locationGorup) {
                                            WhereScriptDTO wsDTO = new WhereScriptDTO();
                                            int endDetweenDay = 15;
                                            String endTime = DateUtil.format(DateUtil.offset(DateUtil.date(), DateField.DAY_OF_MONTH, endDetweenDay), matching.getTimeFieldValue());
                                            String staTime =  DateUtil.format(DateUtil.offset(DateUtil.date(), DateField.DAY_OF_MONTH, -betweenDay), matching.getTimeFieldValue());
                                            String whereSql = StrUtil.replace(StrUtil.replace(kv.getSourceKey(),"="," = '"),kv.getSourceValue(),"cast( "+ kv.getSourceValue() +" as string )") + "' and "+"cast( "+ kv.getTimeKey() +" as int )"+ " BETWEEN " + staTime +" and "+ endTime;
                                            wsDTO.setSourceDbName(s.getTblDatabaseName());
                                            wsDTO.setSourceTableName(s.getTblName());
                                            wsDTO.setTargetDbName(prefix+s.getTblDatabaseName());
                                            wsDTO.setTargetTableName(s.getTblName());
                                            wsDTO.setSourceQueryCriteria(whereSql);
                                            wsDTO.setTargetQueryCriteria(wsDTO.getSourceQueryCriteria());
                                            log.warn("==========================【源分区多种value值】当前数据比对条件"+wsDTO.getSourceQueryCriteria());
                                            whereScriptDTOS.add(wsDTO);
                                        }
                                    }
                                }else{
                                    whereJH(jQCluster,wGCluster,s,prefix,syncRule,ruleToDistCPDTO,whereScriptDTOS,syncRuleConfSet,hdfsHeatBackups);
                                }
                            }
                        }
                    }else{
                        whereJH(jQCluster,wGCluster,s,prefix,syncRule,ruleToDistCPDTO,whereScriptDTOS,syncRuleConfSet,hdfsHeatBackups);
                    }
                });
                //每个规则匹配后异步修改状态
                try {
                    asyncTask.updateToSyncType(syncRuleDTO);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        //将比对sql对象 赋值给传输DTO
        ruleToDistCPDTO.setSyncRuleConfSet(syncRuleConfSet);
        log.warn("==================="+syncRuleConfSet.toString());
        ruleToDistCPDTO.setWhereScriptDTOS(whereScriptDTOS.stream().distinct().collect(Collectors.toList()));
        ruleToDistCPDTO.setMessageId(id);
        ruleToDistCPDTO.setHdfsHeatBackups(hdfsHeatBackups.stream().distinct().collect(Collectors.toList()));
        String s  = "";
        //不为空且的等于“是”
        if(StrUtil.isNotBlank(cpWhether)&&"是".equals(cpWhether)){
            if(ObjectUtil.isNotEmpty(ruleToDistCPDTO.getHdfsHeatBackups())) {
                if (ruleToDistCPDTO.getHdfsHeatBackups().size() > 0) {
                    s = ruleToDistCPService.heatBackup(ruleToDistCPDTO,logId);
                }else{
                    log.warn("===================HDFS路径集合为空");
                }
            }else{
                log.warn("===================HDFS路径集合为空");
            }
        }
        //推送调度
        xxJobLogService.messageToXxlJob(id);
        if(StrUtil.isBlank(s)){
            return true;
        }else{
            return false;
        }
    }

    /**
     * 生成同步规则(文档)
     * @return
     */
    @Override
    public Boolean getSyncRuleDoc(String source) {
        //查询默认规则
        LambdaQueryWrapper<SyncRuleConf> syncRuleWrapper = Wrappers.lambdaQuery();
        syncRuleWrapper.eq(SyncRuleConf::getDel, CommonConstants.STATUS_NORMAL)
                .eq(SyncRuleConf::getIsDefaultRule, ScanCommonConstants.ClusterName.TRUE.value)
                .eq(SyncRuleConf::getExeFrequency, ScanCommonConstants.ClusterName.ONE.value);
        List<SyncRuleConf> syncRuleConfs = this.baseMapper.selectList(syncRuleWrapper);
        //拼接查询源数据表sql
        syncRuleConfs.forEach(syncRule -> {
            ScanSyncRuleDTO syncRuleDTO = splQuSoeDataTable(syncRule);
            List<ScanMetadata> scanMetadataList = scanMetadataService.qyScanMetadataWhereWord(syncRuleDTO);
            //判断规则是否覆盖
            if(scanMetadataList!=null&&scanMetadataList.size()>0){
                syncRule.setExeFrequency(0);
                this.updateById(syncRule);
            }
            //判断数据类型 创建对应文档
            dataType(scanMetadataList,syncRuleDTO);
        });
        return true;
    }

    /**
     * 拼接查询源数据表所需条件1
     * @param syncRule
     * @return
     */
    public ScanSyncRuleDTO splQuSoeDataTable(SyncRuleConf syncRule){
        ScanSyncRuleDTO syncRuleDTO = new ScanSyncRuleDTO();
        //复制
        BeanUtil.copyProperties(syncRule, syncRuleDTO);
//        Date today = DateUtil.date();
//        Date yesterday = DateUtil.yesterday();
//        syncRuleDTO.setStaTime(today);
//        syncRuleDTO.setEndTime(yesterday);
        //字符切割
        syncRuleDTO.setBaseNameWhiteList(StrSpliter.split(syncRule.getBaseNameWhite(),StrUtil.COMMA, true, true));
        syncRuleDTO.setBaseNameBlackList(StrSpliter.split(syncRule.getBaseNameBlack(),StrUtil.COMMA, true, true));
        syncRuleDTO.setTableNameWhiteList(StrSpliter.split(syncRule.getTableNameWhite(),StrUtil.COMMA, true, true));
        syncRuleDTO.setTableNameBlackList(StrSpliter.split(syncRule.getTableNameBlack(),StrUtil.COMMA, true, true));
        syncRuleDTO.setPartNameWhiteList(StrSpliter.split(syncRule.getPartNameWhite(),StrUtil.COMMA, true, true));
        syncRuleDTO.setPartNameBlackList(StrSpliter.split(syncRule.getPartNameBlack(),StrUtil.COMMA, true, true));
        syncRuleDTO.setPartValueWhiteList(StrSpliter.split(syncRule.getPartValueWhite(),StrUtil.COMMA, true, true));
        syncRuleDTO.setPartValueBlackList(StrSpliter.split(syncRule.getPartValueBlack(),StrUtil.COMMA, true, true));
        //时间分区判断
        List<String> fieldMatchings = new ArrayList<>();
        if("分区".equals(syncRule.getIsPart())){
            LambdaQueryWrapper<TimeFieldMatching> timeFielWrapper = Wrappers.lambdaQuery();
            timeFielWrapper.select(TimeFieldMatching::getTimeFieldKay).eq(TimeFieldMatching::getDel,CommonConstants.STATUS_NORMAL);
            fieldMatchings = (List<String>)(List) timeFieldMatchingService.listObjs(timeFielWrapper);
            if("有时间分区".equals(syncRule.getIsTimePart())){
                syncRuleDTO.setPartTimeWhiteList(fieldMatchings);
            }
            if("无时间分区".equals(syncRule.getIsTimePart())){
                syncRuleDTO.setPartTimeBlackList(fieldMatchings);
            }
        }
        return syncRuleDTO;
    }

    /**
     * 生成规则文件
     * @param dto
     * @return
     */
    public void dataType(List<ScanMetadata> scanMetadataList,ScanSyncRuleDTO dto){
        hiveToOracleService.txtFileSD(scanMetadataList,dto.getSyncRuleName());
    }

    /**
     * 通过规则生成查询接口
     * @param syncRule
     * @return
     */
    public String whereLike(SyncRuleConf syncRule){
        String whereLikeWhiteTOP = "and (";
        String whereLikeBlackTOP = "and not (";
        String whereLikeLO = " )";
        String whereLikeWhite = "";
        String whereLikeBlack = "";
        List<String> whereWhiteList  = StrSpliter.split(syncRule.getPartValueWhite(),StrUtil.COMMA,true,true);
        List<String> whereBlackList  = StrSpliter.split(syncRule.getPartValueBlack(),StrUtil.COMMA,true,true);
        //白名单
        if(whereWhiteList!=null&&whereWhiteList.size()>0){
            for (String wh: whereWhiteList) {
                if(StrUtil.isBlank(whereLikeWhite)){
                    whereLikeWhite = "p.part_name like '%" + wh + "%'";
                }else{
                    whereLikeWhite += " or p.part_name like '%" + wh + "%'";
                }
            }
            whereLikeWhite = whereLikeWhiteTOP + whereLikeWhite + whereLikeLO;
        }
        //黑名单
        if(whereBlackList!=null&&whereBlackList.size()>0){
            for (String wh: whereBlackList) {
                if(StrUtil.isBlank(whereLikeBlack)){
                    whereLikeBlack = "p.part_name like '%" + wh + "%'";
                }else{
                    whereLikeBlack += " or p.part_name like '%" + wh + "%'";
                }
            }
            whereLikeBlack = whereLikeBlackTOP + whereLikeBlack + whereLikeLO;
        }
        return whereLikeWhite + whereLikeBlack;
    }


    /**
     * 数据对比所需条件拼接查询impala
     * cast( 源分区字段 as string )='1' and cast( 时间分区字段 as int ) BETWEEN 20231001 and 20231101
     * @param locationList
     * @return
     */
    public WhereScriptDTO whereScript(List<String> locationList,String tblDatabaseName,String tblName, TimeFieldMatching matching, String syncMode, int betweenDay,String prefix){
        String where = "";
        //时间区间
        String staTime = "";
        String endTime = "";
        if(ObjectUtils.isNotEmpty(locationList)){
            if(ObjectUtils.isNotEmpty(locationList.get(0).split(tblName+"/"))&&locationList.get(0).split(tblName+"/").length==2){
                String partNameXX = locationList.get(0).split(tblName+"/")[1];
                if(ObjectUtils.isNotEmpty(partNameXX.split(matching.getTimeFieldKay()+"="))&& partNameXX.split(matching.getTimeFieldKay()+"=").length>=2){
//                    endTime = partNameXX.split(matching.getTimeFieldKay()+"=")[1].split("/")[0];
//                    if(!isValidDateFormat(endTime,matching.getTimeFieldValue())){
//                        endTime = DateUtil.format(DateUtil.date(), matching.getTimeFieldValue());
                        int endDetweenDay = 15;
                        endTime = DateUtil.format(DateUtil.offset(DateUtil.date(), DateField.DAY_OF_MONTH, endDetweenDay), matching.getTimeFieldValue());
//                    }
//                    staTime = locationList.get(locationList.size()-1).split(tblName+"/")[1].split(matching.getTimeFieldKay()+"=")[1].split("/")[0];
//                    if(!isValidDateFormat(staTime,matching.getTimeFieldValue())){
                        staTime =  DateUtil.format(DateUtil.offset(DateUtil.date(), DateField.DAY_OF_MONTH, betweenDay), matching.getTimeFieldValue());
//                    }
                    //判断时间分区在前在后 切割源标识
                    String[] names = partNameXX.split("/");
                    if(ObjectUtils.isNotEmpty(names)){
                        //只有时间分区的
                        if(names.length==1){
                            String[] partNames = names[0].split("=");
                            where = "cast("+ partNames[0] +" as int )"+ " BETWEEN " + staTime +" and "+ endTime;
                        }else {
                            String[] partNamesQ = names[0].split("=");
                            String[] partNamesH = names[1].split("=");
                            if(!"时间窗口同步".equals(syncMode)){
                                //分区在前
                                if(matching.getTimeFieldKay().equals(partNamesQ[0])){
                                    where = StrUtil.replace(StrUtil.replace(names[1],"="," = '"),partNamesH[0],"cast( "+ partNamesH[0] +" as string )") + "' and " + "cast( "+ partNamesQ[0] +" as int )" + " BETWEEN " + staTime +" and "+ endTime;
                                    //分区在后
                                }else if(matching.getTimeFieldKay().equals(partNamesH[0])){
                                    where = StrUtil.replace(StrUtil.replace(names[0],"="," = '"),partNamesQ[0],"cast( "+ partNamesQ[0] +" as string )") + "' and " + "cast( "+ partNamesH[0] +" as int )" + " BETWEEN " + staTime +" and "+ endTime;
                                }
                            }else{
                                //分区在前
                                if(matching.getTimeFieldKay().equals(partNamesQ[0])){
                                    where = "cast("+ partNamesQ[0] +" as int )"+ " BETWEEN " + staTime +" and "+ endTime;
                                    //分区在后
                                }else if(matching.getTimeFieldKay().equals(partNamesH[0])){
                                    where = "cast("+ partNamesH[0] +" as int )"+ " BETWEEN " + staTime +" and "+ endTime;
                                }
                            }

                        }
                    }
                }
            }
        }
        WhereScriptDTO whereScriptDTO = new WhereScriptDTO();
        whereScriptDTO.setSourceDbName(tblDatabaseName);
        whereScriptDTO.setSourceTableName(tblName);
        whereScriptDTO.setTargetDbName(prefix+tblDatabaseName);
        whereScriptDTO.setTargetTableName(tblName);
        whereScriptDTO.setSourceQueryCriteria(where);
        whereScriptDTO.setTargetQueryCriteria(whereScriptDTO.getSourceQueryCriteria());
//        log.warn("==========================当前数据比对条件"+whereScriptDTO.getSourceQueryCriteria());
        return whereScriptDTO ;
    }

    /**
     * 判断字符串是否为指定格式的日期时间
     *
     * @param dateStr    需要检查的日期字符串
     * @param dateFormat 指定的日期格式，例如："yyyyMMdd", "yyyy-MM-dd", "yyyy/MM/dd" 等
     * @return 如果字符串是指定格式的日期时间，返回 true;否则返回 false。
     */

    public static boolean isValidDateFormat(String dateStr, String dateFormat) {
        if (ObjectUtil.isEmpty(dateStr)) {
            return false;
        }
        try {
            DateUtil.parse(dateStr, dateFormat); // 将字符串解析为日期对象，如果解析成功，则说明字符串是有效的日期格式；否则说明字符串不是有效的日期格式。
            return true;
        } catch (Exception e) {
            return false;
        }
    }


    /**
     * 查询不到hdfs路径时，获取规则表信息进行稽核
     * @param jQCluster 金桥
     * @param wGCluster 万国
     * @param jQCluster
     * @param wGCluster
     * @param s
     * @param prefix
     * @param syncRule
     * @param ruleToDistCPDTO
     * @param whereScriptDTOS
     * @param syncRuleConfSet
     * @param hdfsHeatBackups
     */
    public void whereJH(ClusterSparingConf jQCluster,ClusterSparingConf wGCluster,ScanMetadata s,String prefix,SyncRuleConf syncRule,RuleToDistCPDTO ruleToDistCPDTO,List<WhereScriptDTO> whereScriptDTOS,Set<String> syncRuleConfSet,List<HdfsHeatBackup> hdfsHeatBackups){
        HdfsHeatBackup kup = new HdfsHeatBackup();
        kup.setSourceAddress(jQCluster.getHdfsPrefixPath()+s.getTblDatabaseName()+".db/"+s.getTblName());
        kup.setTargetAddress(wGCluster.getHdfsPrefixPath()+prefix+s.getTblDatabaseName()+".db/"+s.getTblName());
        kup.setSourceDbName(s.getTblDatabaseName());
        kup.setTargetDbName(prefix+s.getTblDatabaseName());
        kup.setSourceTableName(s.getTblName());
        kup.setTargetTableName(s.getTblName());
        if (StrUtil.isNotBlank(syncRule.getSourceName())) {
            kup.setSourceData (syncRule.getSourceName());
            ruleToDistCPDTO.setSourceDataName(syncRule.getSourceName());
            kup.setBackupSort("1");
            ruleToDistCPDTO.setBackupSort("1");
        } else {
            kup.setSourceData("无源");
            ruleToDistCPDTO.setSourceDataName("无源");
            kup.setBackupSort("1");
            ruleToDistCPDTO.setBackupSort("1");
        }
        hdfsHeatBackups.add(kup);
        if(ObjectUtil.isNotEmpty(syncRuleConfSet)){
            for (String partName: syncRuleConfSet) {
                WhereScriptDTO whereScriptDTO = new WhereScriptDTO();
                //当有源分区时对比条件拼接要加上源分区
                whereScriptDTO.setSourceDbName(s.getTblDatabaseName());
                whereScriptDTO.setSourceTableName(s.getTblName());
                if(StrUtil.isNotBlank(partName)){
                    String[] partNames = partName.split("=");
                    whereScriptDTO.setSourceQueryCriteria(StrUtil.replace(StrUtil.replace(partName,"=","='"),partNames[0],"cast( "+ partNames[0] +" as string )") + "'");
                }else{
                    whereScriptDTO.setSourceQueryCriteria("");
                }
                whereScriptDTO.setTargetQueryCriteria(whereScriptDTO.getSourceQueryCriteria());
                whereScriptDTO.setTargetTableName(s.getTblName());
                whereScriptDTO.setTargetDbName(prefix+s.getTblDatabaseName());
                whereScriptDTOS.add(whereScriptDTO);
                log.warn("==========================【全量源标识】当前数据比对条件"+whereScriptDTO.getSourceQueryCriteria());
            }
        }else{
            WhereScriptDTO whereScriptDTO = new WhereScriptDTO();
            //当有源分区时对比条件拼接要加上源分区
            whereScriptDTO.setSourceDbName(s.getTblDatabaseName());
            whereScriptDTO.setSourceTableName(s.getTblName());
            whereScriptDTO.setSourceQueryCriteria("");
            whereScriptDTO.setTargetQueryCriteria(whereScriptDTO.getSourceQueryCriteria());
            whereScriptDTO.setTargetTableName(s.getTblName());
            whereScriptDTO.setTargetDbName(prefix+s.getTblDatabaseName());
            whereScriptDTOS.add(whereScriptDTO);
            log.warn("==========================【全量源标识】当前数据比对条件"+whereScriptDTO.getSourceQueryCriteria());
        }
    }

    @Override
    public List<HotStandbyLogSource> sourceHotBackupState() {
        //源级热备同步
        LambdaQueryWrapper<SyncRuleConf> wrapper = Wrappers.lambdaQuery();
        wrapper.select(SyncRuleConf::getSourceName)
               .groupBy(SyncRuleConf::getSourceName);
        List<SyncRuleConf> resultList = syncRuleConfMapper.selectList(wrapper);
        List<HotStandbyLogSource> sourceHotBackupState = hotStandbyLogSourceService.sourceHotBackupState();
        Map<String, HotStandbyLogSource> map = sourceHotBackupState.stream()
                .collect(Collectors.toMap(HotStandbyLogSource::getSourceName, entity -> entity));
        for (SyncRuleConf syncRuleConf : resultList) {
            if (syncRuleConf == null){
                boolean strTar = map.containsKey("无源");
                if (!strTar){
                    HotStandbyLogSource hotStandbyLogSource = new HotStandbyLogSource();
                    hotStandbyLogSource.setSourceName("无源");
                    hotStandbyLogSource.setTaskState("待执行");
                    sourceHotBackupState.add(hotStandbyLogSource);
                }
            }else {
                boolean strTar = map.containsKey(syncRuleConf.getSourceName());
                if (!strTar){
                    HotStandbyLogSource hotStandbyLogSource = new HotStandbyLogSource();
                    hotStandbyLogSource.setSourceName(syncRuleConf.getSourceName());
                    hotStandbyLogSource.setTaskState("待执行");
                    sourceHotBackupState.add(hotStandbyLogSource);
                }
            }
        }
        //ty的库级同步
        String[] tyDb = {"ty_dwm","ty_dwd","ty_ods"};
        List<SyncCompleteLibraryTask> syncCompleteLibraryTasks = syncCompleteLibraryTaskService.dbHotBackupState();
        Map<String, SyncCompleteLibraryTask> dbMap = syncCompleteLibraryTasks.stream()
                .collect(Collectors.toMap(SyncCompleteLibraryTask::getDatabaseName, entity -> entity));
        for (String s : tyDb) {
            boolean strTar = dbMap.containsKey(s);
            if (strTar){
                SyncCompleteLibraryTask syncCompleteLibraryTask = dbMap.get(s);
                HotStandbyLogSource hotStandbyLogSource = new HotStandbyLogSource();
                hotStandbyLogSource.setSourceName(syncCompleteLibraryTask.getDatabaseName());
                hotStandbyLogSource.setTaskState(syncCompleteLibraryTask.getTaskState());
                sourceHotBackupState.add(hotStandbyLogSource);
            }else {
                HotStandbyLogSource hotStandbyLogSource = new HotStandbyLogSource();
                hotStandbyLogSource.setSourceName(s);
                hotStandbyLogSource.setTaskState("待执行");
                sourceHotBackupState.add(hotStandbyLogSource);
            }
        }
        return sourceHotBackupState;
    }
}
